package org.pizza.did.zookeeper;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.pizza.did.api.WorkerIdDistributor;
import org.pizza.did.zookeeper.exception.CheckLastTimeException;
import org.pizza.did.zookeeper.exception.MaxWorkIdException;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.commons.io.FileUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLBackgroundPathAndBytesable;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZookeeperDistributor implements WorkerIdDistributor {
    private static final Logger log = LoggerFactory.getLogger(ZookeeperDistributor.class);

    private final String applicationName;

    private String zkAddressNode = null;

    private String listenAddress;

    private long workerId;

    private long maxWorkId;

    private String localCachePath;

    private String zkForeverPath;

    private String ip;

    private String port;

    private String zkConnectionString;

    private long lastUpdateTime;

    private CuratorFramework curator;

    public ZookeeperDistributor(String applicationName, String ip, String port, String zkConnectionString, long maxWorkId) {
        this.ip = ip;
        this.port = port;
        this.maxWorkId = maxWorkId;
        this.applicationName = applicationName;
        this.listenAddress = ip + ":" + port;
        this.zkConnectionString = zkConnectionString;
        this.localCachePath = System.getProperty("java.io.tmpdir") + File.separator + applicationName + "/did/{port}/workerID.properties";
        this.zkForeverPath = "/snowflake/" + applicationName + "/forever";
    }

    public boolean init() {
        return true;
    }

    private boolean get() {
        try {
            log.info("init applicationName:{} ip:{} port:{} zkAddress:{} ", this.applicationName, this.ip, this.port, this.zkConnectionString);
            openCurator(this.zkConnectionString, (RetryPolicy) new RetryUntilElapsed(1000, 4), 12000);
            Stat stat = (Stat) this.curator.checkExists().forPath(this.zkForeverPath);
            if (stat == null) {
                onNothingness();
            } else {
                onExisting();
            }
        } catch (Exception e) {
            log.error("start node error ", e);
            try {
                Properties properties = new Properties();
                properties.load(new FileInputStream(new File(this.localCachePath.replace("{port}", this.port + ""))));
                this.workerId = Integer.parseInt(properties.getProperty("workerId"));
                log.warn("start failed ,use local node file properties workerId-{}", this.workerId);
            } catch (Exception e1) {
                log.error("read file error ", e1);
                return true;
            }
        }
        return false;
    }

    public long distribute() {
        if (get()) {
            return getWorkerId();
        }
        throw new RuntimeException("分配workId失败");
    }

    private void onExisting() throws Exception {
        Map<String, Integer> nodeMap = new HashMap<>();
        Map<String, String> realNode = new HashMap<>();
        //查询所有子节点
        List<String> keys = this.curator.getChildren().forPath(this.zkForeverPath);
        for (String key : keys) {
            String[] nodeKey = key.split("-");
            realNode.put(nodeKey[0], key);
            nodeMap.put(nodeKey[0], Integer.parseInt(nodeKey[nodeKey.length - 1]));
        }
        //子节点总数大于最大id分配数
        if (keys.size() > this.maxWorkId) {
            throw new MaxWorkIdException("exceed max workId limit");
        }
        //获取当前地址listenAddress
        Integer workerId = nodeMap.get(this.listenAddress);
        //存在workerId(当前listenAddress有服务在维护)
        if (workerId != null) {
            this.zkAddressNode = this.zkForeverPath + "/" + realNode.get(this.listenAddress);
            this.workerId = workerId;
            //如果节点数据的上报 大于 当前时间
            if (checkTimeStamp()) {
                throw new CheckLastTimeException("init timestamp check error,forever node timestamp gt this current time");
            }
            scheduledUploadData();
            updateLocalWorkerId();
            log.info("[{} old node] find forever node have this endpoint ip-{} port-{} workId-{} childNode and start success", this.applicationName, this.ip, this.port, Long.valueOf(this.workerId));
        } else {
            String newNode = createNode();
            this.zkAddressNode = newNode;
            String[] nodeKey = newNode.split("-");
            this.workerId = Integer.parseInt(nodeKey[nodeKey.length - 1]);
            scheduledUploadData();
            updateLocalWorkerId();
            log.info("[{} new node] can not find node on forever node that endpoint ip-{} port-{} workId-{},create own node on forever node and start success ", new Object[]{this.applicationName, this.ip, this.port, Long.valueOf(this.workerId)});
        }
    }

    private void onNothingness() throws Exception {
        this.zkAddressNode = createNode();
        updateLocalWorkerId();
        scheduledUploadData();
    }

    private void scheduledUploadData() {
        Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = new Thread(r, "ZookeeperDistributor-");
            thread.setDaemon(true);
            return thread;
        })
                //每60s上报数据
                .scheduleWithFixedDelay(this::updateNewData, 1L, 60L, TimeUnit.SECONDS);
    }

    private boolean checkTimeStamp() throws Exception {
        byte[] bytes = this.curator.getData().forPath(this.zkAddressNode);
        Endpoint endPoint = deBuildData(new String(bytes));
        //上次上报的时间 小于 当前时间
        return endPoint.getTimestamp() > System.currentTimeMillis();
    }

    private String createNode() throws Exception {
        try {
            return (String) ((ACLBackgroundPathAndBytesable) this.curator.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT_SEQUENTIAL))
                    .forPath(this.zkForeverPath + "/" + this.listenAddress + "-", buildData().getBytes());
        } catch (Exception e) {
            log.error("create node error msg {} ", e.getMessage());
            throw e;
        }
    }

    private void updateNewData() {
        try {
            if (System.currentTimeMillis() < this.lastUpdateTime) {
                //已经上报过了
                return;
            }
            //更新数据
            this.curator.setData().forPath(this.zkAddressNode, buildData().getBytes());
            this.lastUpdateTime = System.currentTimeMillis();
        } catch (Exception e) {
            log.info("update init data error path is error, this.zkAddressNode", e);
        }
    }

    private String buildData() throws JsonProcessingException {
        Endpoint endpoint = new Endpoint(this.ip, this.port, System.currentTimeMillis());
        ObjectMapper mapper = new ObjectMapper();
        return mapper.writeValueAsString(endpoint);
    }

    private Endpoint deBuildData(String json) throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        return (Endpoint) mapper.readValue(json, Endpoint.class);
    }

    private void updateLocalWorkerId() {
        File leafConfFile = new File(this.localCachePath.replace("{port}", this.port));
        boolean exists = leafConfFile.exists();
        log.info("file exists status is {}", exists);
        if (exists) {
            try {
                FileUtils.writeStringToFile(leafConfFile, "workerId=" + this.workerId, false);
                log.info("update file cache workerId is {}", this.workerId);
            } catch (IOException e) {
                log.error("update file cache error ", e);
            }
        } else {
            try {
                boolean mkdirs = leafConfFile.getParentFile().mkdirs();
                log.info("init local file cache create parent dis status is {}, worker id is {}", mkdirs, this.workerId);
                if (mkdirs) {
                    if (leafConfFile.createNewFile()) {
                        FileUtils.writeStringToFile(leafConfFile, "workerId=" + this.workerId, false);
                        log.info("local file cache workerID is {}", this.workerId);
                    }
                } else {
                    log.warn("create parent dir error===");
                }
            } catch (IOException e) {
                log.warn("create workerID conf file error", e);
            }
        }
    }

    private void openCurator(String connectionString, RetryPolicy retryPolicy, int sessionTimeoutMs) {
        this.curator = CuratorFrameworkFactory.builder().connectString(connectionString).retryPolicy(retryPolicy).connectionTimeoutMs(10000).sessionTimeoutMs(sessionTimeoutMs).build();
        this.curator.start();
    }

    static class Endpoint {
        private String ip;

        private String port;
        //上报的时间
        private long timestamp;

        public Endpoint() {
        }

        public Endpoint(String ip, String port, long timestamp) {
            this.ip = ip;
            this.port = port;
            this.timestamp = timestamp;
        }

        public String getIp() {
            return this.ip;
        }

        public void setIp(String ip) {
            this.ip = ip;
        }

        public String getPort() {
            return this.port;
        }

        public void setPort(String port) {
            this.port = port;
        }

        public long getTimestamp() {
            return this.timestamp;
        }

        public void setTimestamp(long timestamp) {
            this.timestamp = timestamp;
        }
    }

    public String getZkAddressNode() {
        return this.zkAddressNode;
    }

    public void setZkAddressNode(String zkAddressNode) {
        this.zkAddressNode = zkAddressNode;
    }

    public String getListenAddress() {
        return this.listenAddress;
    }

    public void setListenAddress(String listenAddress) {
        this.listenAddress = listenAddress;
    }

    public long getWorkerId() {
        return this.workerId;
    }

    public void setWorkerId(long workerId) {
        this.workerId = workerId;
    }

    public static void main(String[] args) {
        ZookeeperDistributor holder1 = new ZookeeperDistributor("seckill", "127.0.0.1", "8000", "127.0.0.1:2181", 31L);
        holder1.init();
    }
}
